package com.fwmagic.akka.rpc

import akka.actor.{Actor, ActorSystem, Props}
import com.typesafe.config.{Config, ConfigFactory}
import java.util.concurrent.TimeUnit
import scala.collection.mutable.HashMap
import scala.concurrent.duration.FiniteDuration

/**
  * 接收消息，管理Workers:Worker注册和移除长时间不发送心跳的Worker
  */
class Master extends Actor {

    //存储所有的的Worker信息
    val id2Worker = new HashMap[String, WorkerInfo]()

    /**
      * Master启动后再启动一个定时器，用于检测超时的Worker(根据Worker的lastUpdateTime),然后去除
      */
    override def preStart(): Unit = {

        import context.dispatcher

        context.system.scheduler.schedule(
            new FiniteDuration(0, TimeUnit.SECONDS),
            new FiniteDuration(10, TimeUnit.SECONDS),
            self,
            CheckTimeOutWorker)
    }

    /**
      * 接收消息
      *
      * @return
      */
    override def receive: Receive = {
        //Worker发送给Master的注册消息
        case RegisterWorker(workerId, memory, cores) => {
            //数据封装起来
            val workerInfo = new WorkerInfo(workerId, memory, cores)

            println("===>Receive WorkInfo:" + workerInfo)

            //保存到内存中
            id2Worker(workerId) = workerInfo

            //Master向Worker发送一个注册成功的消息
            sender() ! RegistedWorker
        }

        //Worker报活消息，需更新worker的最后报活时间(lastUpdateTime)
        case HeartBeat(workerId) => {
            println(System.currentTimeMillis() + "-" + workerId + " is alive!")
            //根据workerId取id2Worker中到对应的WorkInfo
            if (id2Worker.contains(workerId)) {
                //取出workerInfo
                val workerInfo = id2Worker(workerId)
                //更新workerInfo最近一次心跳时间
                workerInfo.lastUpdateTime = System.currentTimeMillis()
            }
        }

        //接收消息，检测并移除超时的Worker
        case CheckTimeOutWorker => {
            val workers: Iterable[WorkerInfo] = id2Worker.values
            val deadWorkers: Iterable[WorkerInfo] = workers.filter(w => System.currentTimeMillis() - w.lastUpdateTime > 10000)
            deadWorkers.foreach(dw => id2Worker -= dw.id)
            println(System.currentTimeMillis() + "- current alive workers:" + id2Worker.size)
        }
    }
}

/**
  * 创建masterActorSystem和masterActor
  */
object Master {
    def main(args: Array[String]): Unit = {
        val host = "localhost"
        val port = "8888"
        //定义配置文件
        val configStr =
            s"""
               |akka.actor.provider="akka.remote.RemoteActorRefProvider"
               |akka.remote.netty.tcp.hostname=$host
               |akka.remote.netty.tcp.port=$port
               |akka.remote.artery.enabled=false
            """.stripMargin

        //解析配置文件，生成Config对象
        val config: Config = ConfigFactory.parseString(configStr)

        //创建ActorSystem
        val masterActorSystem: ActorSystem = ActorSystem("Master-Actor-System", config)

        //创建Actor
        //val masterActor: ActorRef = masterActorSystem.actorOf(Props[Master], "Master-Actor")
        masterActorSystem.actorOf(Props[Master], "Master-Actor")

        //（异步的）发送消息给Master
        //masterActor.!("hello")

        // masterActor ! "hi"
    }
}
